package org.rrd.kafka.phoneix.db;

import com.alibaba.otter.canal.protocol.FlatMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.rrd.kafka.phoneix.entity.TestToHbase;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * @program: dataStreming
 * @description:
 * @author: sunteng
 * @create: 2020-02-27 09:37
 **/
@Slf4j
public class PhoneixJdbc extends RichSinkFunction<FlatMessage> {

    private Connection connection;
    private PreparedStatement upsertPreparedStatement;
    private PreparedStatement deletePreparedStatement;

    @Override
    public void open(Configuration parameters) throws Exception {
        // JDBC连接信息

        String DRIVERNAME = "org.apache.phoenix.jdbc.PhoenixDriver";
//        String DBURL = "jdbc:phoenix:10.10.12.215,10.10.12.216,10.10.12.129:2181:/hbase";
        String DBURL = "jdbc:phoenix:172.16.1.90,172.16.1.91,172.16.2.148:2181:/hbase";
        // 加载JDBC驱动
        Class.forName(DRIVERNAME);
        // 获取数据库连接
        connection = DriverManager.getConnection(DBURL);
        connection.setAutoCommit(true);


        String upsertSql = "upsert into \"testToHbase\" values(?, ?, ?, ?, ?, ?)";
        upsertPreparedStatement = connection.prepareStatement(upsertSql);

        String deleteSql = "delete from \"testToHbase\" where \"id\" = ?";
        deletePreparedStatement = connection.prepareStatement(deleteSql);


        super.open(parameters);
    }

    @Override
    public void close() throws Exception {
        upsertPreparedStatement.close();
        deletePreparedStatement.close();
        log.info("##close");
        super.close();
    }

    @Override
    public void invoke(FlatMessage flatMessage, Context context) throws Exception {
        ObjectMapper mapper = new ObjectMapper();

        log.info("##begin deal message");
        TestToHbase testToHbase = mapper.convertValue(flatMessage.getData().get(0), TestToHbase.class);

        log.info("## is DML");
        if (flatMessage.getType().equals("DELETE")) {
            log.info("##DELETE");
            deletePreparedStatement.setLong(1, testToHbase.getId());
            deletePreparedStatement.execute();

        } else if (flatMessage.getType().equals("INSERT") || flatMessage.getType().equals("UPDATE")) {
            log.info("##UPSERT {}", testToHbase.getId() + testToHbase.getName());
            upsertPreparedStatement.setLong(1, testToHbase.getId());
            upsertPreparedStatement.setString(2, testToHbase.getName());
            upsertPreparedStatement.setInt(3, testToHbase.getCreate_time());
            upsertPreparedStatement.setInt(4, testToHbase.getStatus());
            upsertPreparedStatement.setFloat(5, testToHbase.getAmount());
            upsertPreparedStatement.setString(6, testToHbase.getBlob());
            upsertPreparedStatement.executeUpdate();
        }
    }

}
